package com.ry.flink.jobx;

import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStream;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Schema;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

/**
 * 2. **Append-only 流（insert-only）：** 仅通过 `INSERT` 操作修改的动态表可以通过输出插入的行转换为流。
 *
 * */
public class Job2 {
    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        StreamTableEnvironment tablEnv = StreamTableEnvironment.create(env);

        DataStreamSource<String> source = env.fromElements("貂蝉", "孙尚香", "大乔和小乔");

        Table table = tablEnv.fromDataStream(source);

        tablEnv.createTemporaryView("dream",source);

        Table tableResult = tablEnv.sqlQuery("select * from dream");

        DataStream<Row> rowDataStream = tablEnv.toDataStream(tableResult);
        /**
         *  +I[孙尚香]
         *  +I[大乔和小乔]
         *  +I[貂蝉]
         * */
        rowDataStream.print();
        env.execute("Job2");
    }
}
